package org.test.service;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.service.CompositeService;
import org.apache.hadoop.service.Service;
import org.apache.hadoop.yarn.event.AsyncDispatcher;
import org.apache.hadoop.yarn.event.Dispatcher;
import org.apache.hadoop.yarn.event.EventHandler;

public class SimpleMaster extends CompositeService {

	private Dispatcher dispatcher;
	private String jobId;
	private int taskNum;
	private String[] taskIds;
	
	public SimpleMaster(String name,String jobId, int taskNum) {
		super(name);
		this.jobId = jobId;
		this.taskNum = taskNum;
		taskIds = new String[taskNum];
		for(int i=0;i<taskNum;i++){
			taskIds[i] = this.jobId+"_task_"+i;
		}
	}

	@Override
	protected void serviceInit(Configuration conf) throws Exception {
		
		dispatcher = new AsyncDispatcher();
		dispatcher.register(JobEventType.class, new JobEventHandler());
		dispatcher.register(TaskEventType.class, new TaskEventHandler());
		addService((Service)dispatcher);
		super.serviceInit(conf);
	}

	public Dispatcher getDispatcher() {
		return dispatcher;
	}
	
	
	private class JobEventHandler implements EventHandler<JobEvent>{

		public void handle(JobEvent event) {
			if(event.getType() == JobEventType.BEGIN){
				System.out.println("Job begin:"+event.getJobId());
				for(int i=0;i<taskNum;i++){
					dispatcher.getEventHandler().handle(new TaskEvent(taskIds[i], TaskEventType.BEGIN));
				}
			}else if(event.getType() == JobEventType.END){
				System.out.println("Job end:"+event.getJobId());
				for(int i=0;i<taskNum;i++){
					dispatcher.getEventHandler().handle(new TaskEvent(taskIds[i], TaskEventType.END));
				}
			}
		}
		
	}
	
	
	private class TaskEventHandler implements EventHandler<TaskEvent>{

		public void handle(TaskEvent event) {
			if(event.getType() == TaskEventType.BEGIN){
				System.out.println("task begin:"+event.getTaskId());
			}else if(event.getType() == TaskEventType.END){
				System.out.println("Task end:"+event.getTaskId());
			}
		}
		
	}

}
